﻿
using System;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace WalkthroughTDF
{
    static class TestActionBlock
    {
        private static void ProcessNumber(int num)
        {
            Thread.Sleep(500);
            Console.WriteLine("thread<{0}>: {1}", Thread.CurrentThread.ManagedThreadId, num);
        }

        private static Task AsyncProcessNumber(int number)
        {
            Console.WriteLine("start processing '{0}'", number);
            return Task.Factory.StartNew((state) =>
                                      {
                                          Thread.Sleep(500);
                                          Console.WriteLine("end processing '{0}'", state);
                                      }, number);
        }

        private static void PostNumbers(ActionBlock<int> ab, int total)
        {
            for (int index = 0; index < total; ++index)
                ab.Post(index);
        }

        /// <summary>
        /// chekanote: data in the queue is processed asynchronously, (by default, executed in a thread from the pool)
        /// but the order is reserved, as the same order they are posted into the queue
        /// </summary>
        private static void SimpleTest()
        {
            ActionBlock<int> ab = new ActionBlock<int>(new Action<int>(ProcessNumber));

            PostNumbers(ab, 10);

            Console.ReadLine();
        }

        /// <summary>
        /// by default, the MaxDegreeOfParallism is 1, that means the data is processed only
        /// by one thread from the pool 
        /// here allow the data is processed in parallel by multiple threads from the pool
        /// chekanote: but the order is still reserved, the data popped out of the queue in the same order as they are inserted
        /// </summary>
        private static void TestMultipleWorkers()
        {
            var blockOpt = new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 };
            ActionBlock<int> ab = new ActionBlock<int>(new Action<int>(ProcessNumber), blockOpt);

            PostNumbers(ab, 10);

            Console.ReadLine();
        }

        private static void TestCompletedTask()
        {
            var blockOpt = new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 3 };
            var ab = new ActionBlock<int>(new Action<int>(ProcessNumber), blockOpt);

            PostNumbers(ab, 10);
            ab.Complete();// notify that there will be no data inserted into that queue
            Console.WriteLine("------------- all data inserted");

            ab.Completion.Wait();// wait until all the data in the queue are processed
            Console.WriteLine("------------- all data processed");
        }

        /// <summary>
        /// process the item retrieved from queue asynchronously
        /// chekanote: although start processing by delegating the work into another task
        /// but fetch another item from the queue is not started when the task is started
        /// but after the runningg task is totally completed
        /// so if there is only one worker for the queue, items still be executed one by one, NOT in parallel
        /// </summary>
        private static void TestAsyncAct()
        {
            var ab = new ActionBlock<int>(new Func<int, Task>(AsyncProcessNumber));

            for (int index = 0; index < 5; index++)
            {
                ab.Post(index);
                // Thread.Sleep(100);
            }
            ab.Complete();
            Console.WriteLine("*********** all inserted");

            ab.Completion.Wait();
            Console.WriteLine("!!!!!!!!!!! all processed");
        }

        public static void TestMain()
        {
            // SimpleTest();
            // TestMultipleWorkers();
            // TestCompletedTask();
            TestAsyncAct();
        }
    }
}
